package org.jetlinks.pro.device.message.writer;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.hswebframework.ezorm.rdb.mapping.ReactiveRepository;
import org.hswebframework.web.crud.events.EntityCreatedEvent;
import org.hswebframework.web.crud.events.EntityDeletedEvent;
import org.hswebframework.web.crud.events.EntityModifyEvent;
import org.hswebframework.web.crud.events.EntitySavedEvent;
import org.jetlinks.core.Values;
import org.jetlinks.core.device.DeviceProductOperator;
import org.jetlinks.core.device.DeviceRegistry;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.message.Headers;
import org.jetlinks.core.metadata.DeviceMetadata;
import org.jetlinks.core.metadata.PropertyMetadata;
import org.jetlinks.core.metadata.types.GeoType;
import org.jetlinks.pro.PropertyConstants;
import org.jetlinks.pro.device.entity.DeviceInstanceEntity;
import org.jetlinks.pro.device.entity.DeviceTagEntity;
import org.jetlinks.pro.device.events.DeviceDeployedEvent;
import org.jetlinks.pro.gateway.DeviceMessageUtils;
import org.jetlinks.pro.gateway.annotation.Subscribe;
import org.jetlinks.pro.geo.GeoObject;
import org.jetlinks.pro.geo.GeoObjectManager;
import org.jetlinks.pro.geo.GeoQueryParameter;
import org.jetlinks.pro.tenant.utils.TenantUtils;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 地理位置数据
 *
 * @author zhouhao
 * @since 1.0
 */
@Component
@Slf4j
@ConditionalOnClass(GeoObject.class)
@ConfigurationProperties(prefix = "device.message.writer.geo")
public class DeviceGeoDataWriter {

    private final GeoObjectManager geoObjectManager;

    private final DeviceRegistry registry;

    private final ReactiveRepository<DeviceTagEntity, String> tagRepository;

    @Getter
    @Setter
    private boolean enableProperty = true;

    public DeviceGeoDataWriter(GeoObjectManager geoObjectManager,
                               DeviceRegistry registry,
                               ReactiveRepository<DeviceTagEntity, String> tagRepository) {
        this.geoObjectManager = geoObjectManager;
        this.registry = registry;
        this.tagRepository = tagRepository;
    }

    private Mono<PropertyMetadata> findGeoProperty(DeviceMessage message, DeviceMetadata metadata) {
        String geoProperty = message.getHeaderOrElse(Headers.geoProperty, null);
        if (null != geoProperty) {
            return Mono.justOrEmpty(metadata.getProperty(geoProperty));
        }
        //遍历查找geo类型的属性,属性较多时,性能稍低
        return Mono
            .justOrEmpty(metadata.findProperty(prop -> prop.getValueType() instanceof GeoType));
    }

    @Subscribe("/device/*/*/message/property/**")
    public Mono<Void> handleDeviceGeoProperty(DeviceMessage msg) {
        if (!enableProperty || !msg.getHeaderOrDefault(Headers.propertyContainsGeo)) {
            return Mono.empty();
        }
        Map<String, Object> properties = DeviceMessageUtils.tryGetProperties(msg).orElse(Collections.emptyMap());

        if (properties.isEmpty()) {
            return Mono.empty();
        }
        return registry
            .getDevice(msg.getDeviceId())
            .flatMap(device -> Mono
                .zip(
                    device.getMetadata()
                          .flatMap(metadata -> findGeoProperty(msg, metadata))
                          .filter(prop -> properties.get(prop.getId()) != null),
                    device.getProduct()
                ))
            .flatMap(devMeta -> {
                PropertyMetadata metadata = devMeta.getT1();
                DeviceProductOperator product = devMeta.getT2();

                GeoObject geoObject = new GeoObject();
                geoObject.setId(msg.getDeviceId().concat(":").concat(metadata.getId()));
                geoObject.setPoint(GeoType.GLOBAL.convert(properties.get(metadata.getId())));
                geoObject.setProperty(metadata.getId());
                geoObject.setObjectId(msg.getDeviceId());
                geoObject.setTimestamp(msg.getTimestamp());
                geoObject.setObjectType("device");
                Map<String, Object> tags = new HashMap<>();
                tags.put("productId", product.getId());
                tags.put("from", "property");
                tags.putAll(TenantUtils.createBindingKey(msg.getHeaders()));
                msg.getHeader(PropertyConstants.deviceName)
                   .ifPresent(str -> tags.put("deviceName", str));

                geoObject.setTags(tags);

                return geoObjectManager.commit(Collections.singleton(geoObject));
            });
    }

    @EventListener
    public void handleDeviceTagEvent(EntityCreatedEvent<DeviceTagEntity> event) {
        event.async(saveTagGeo(event.getEntity()));
    }

    @EventListener
    public void handleDeviceTagEvent(EntitySavedEvent<DeviceTagEntity> event) {
        event.async(saveTagGeo(event.getEntity()));
    }

    @EventListener
    public void handleDeviceTagEvent(EntityDeletedEvent<DeviceTagEntity> event) {
        event.async(
            geoObjectManager
                .remove(event.getEntity()
                             .stream()
                             .map(DeviceTagEntity::getId)
                             .collect(Collectors.toList()))
        );
    }

    @EventListener
    public void handleDeviceEvent(EntityDeletedEvent<DeviceInstanceEntity> event) {
        new GeoQueryParameter()
            .filter(query -> query.in(
                GeoObject::getObjectId,
                event.getEntity()
                     .stream()
                     .map(DeviceInstanceEntity::getId)
                     .collect(Collectors.toList())))
            .as(geoObjectManager::remove)
            .subscribe(
                (r) -> log.debug("delete device geo [{}] objects", r),
                (err) -> log.error("delete device geo objects error", err)
            );
    }

    @EventListener
    public void handleDeviceTagEvent(EntityModifyEvent<DeviceTagEntity> event) {
        event.async(
            saveTagGeo(event.getAfter())
        );
    }

    @EventListener
    public void handleDeviceEvent(EntityModifyEvent<DeviceInstanceEntity> event) {
        event
            .async(
                syncDeviceGeoPoint(event.getAfter())
            );
    }

    @EventListener
    public void handleDeviceEvent(EntitySavedEvent<DeviceInstanceEntity> event) {
        event
            .async(
                syncDeviceGeoPoint(event.getEntity())
            );
    }

    @EventListener
    public void handleDeviceDeployEvent(DeviceDeployedEvent event) {
        event
            .async(
                syncDeviceGeoPoint(event.getDevices())
            );
    }

    Mono<Void> syncDeviceGeoPoint(Collection<DeviceInstanceEntity> devices) {
        Set<String> idSet = devices
            .stream()
            .map(DeviceInstanceEntity::getId)
            .filter(Objects::nonNull)
            .collect(Collectors.toSet());
        if (idSet.isEmpty()) {
            return Mono.empty();
        }
        //同步设备标签中的地理位置信息
        return tagRepository
            .createQuery()
            .in(DeviceTagEntity::getDeviceId, idSet)
            .and(DeviceTagEntity::getType, GeoType.GLOBAL)
            .fetch()
            .as(this::saveTagGeo);
    }

    public Mono<Void> saveTagGeo(Collection<DeviceTagEntity> tagEntities) {
        return saveTagGeo(Flux.fromIterable(tagEntities));
    }

    public Mono<Void> saveTagGeo(Flux<DeviceTagEntity> tagEntities) {
        return tagEntities
            .filter(tag -> GeoType.ID.equals(tag.getType()) || "geo".equals(tag.getType()))
            .flatMap(tag -> {
                GeoObject geoObject = new GeoObject();
                geoObject.setId(tag.getId());
                geoObject.setPoint(GeoType.GLOBAL.convert(tag.getValue()));
                geoObject.setProperty(tag.getKey());
                geoObject.setObjectType("device");
                geoObject.setObjectId(tag.getDeviceId());
                geoObject.setTimestamp(System.currentTimeMillis());
                return registry
                    .getDevice(tag.getDeviceId())
                    .flatMap(device -> device
                        .getSelfConfigs(
                            PropertyConstants.productId.getKey(),
                            PropertyConstants.deviceName.getKey(),
                            PropertyConstants.productName.getKey(),
                            PropertyConstants.tenantId.getKey(),
                            PropertyConstants.tenantMemberId.getKey(),
                            PropertyConstants.bindings.getKey()
                        )
                        .map(Values::getAllValues)
                        .doOnNext(tags -> {
                            Map<String, Object> newTags = new HashMap<>(tags);
                            newTags.putAll(TenantUtils.createBindingKey(tags));
                            newTags.put("from", "tag");
                            geoObject.setTags(newTags);
                        })
                        //为什么在这里return而不在外面?
                        //因为设备未激活时,无法获取到一些重要信息,所以为empty的话就不同步了.
                        .thenReturn(geoObject)
                    );
            })
            .collectList()
            .flatMap(geoObjectManager::commit)
            .doOnError(err -> log.error(err.getMessage(), err))
            .then();
    }


}
